package net.bwie.dws;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class DwsWindowCarsLog {
    public static void main(String[] args) throws Exception {
        // 1.表执行环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取Kafka数据
        createInputTable(tableEnv);

        // 3.数据处理
        Table table = handle(tableEnv);

        tableEnv.createTemporaryView("resultTable", table);
//        tableEnv.executeSql("select * from resultTable").print();

        // 4.映射表 写入doris表中
        createOutputTable(tableEnv);

        // 5.保存数据
        saveToSink(tableEnv, table);


    }

    // todo a.获取表执行环境
    private static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }

    // todo b.创建输入表
    private static void createInputTable(TableEnvironment tableEnv) {
        // 读取汇总表 dwd_summary_cars_log 车辆进出汇总表
        tableEnv.executeSql("CREATE TABLE dwd_summary_cars_log (\n" +
                " id STRING,\n"+
                " opTime STRING,\n" +
                " owerId INT,\n"+
                " ctype STRING,\n"+
                " carCode STRING,\n"+
                " cId BIGINT,\n"+
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(opTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '3' MINUTE" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'dwd_summary_cars_log',\n" +
                " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                " 'properties.group.id' = 'dwd_summary_cars_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n"+
                ")"
        );

//        tableEnv.executeSql("SELECT * FROM  dwd_summary_cars_log").print();
    }

    // todo c.聚合计算
    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "SELECT\n" +
                        // 窗口开始时间
                        "  window_start,\n" +
                        // 窗口结束时间
                        "  window_end,\n" +
                        "  CAST(cId AS INT) AS cId,\n" +
                        "sum(if(ctype = '1' ,1,0)) as in_cnt,\n" +
                        "sum(if(ctype = '2' ,1,0)) as out_cnt\n" +
                        "FROM TABLE(\n" +
                        // 添加 3分钟 的 滚动窗口
                        "  TUMBLE(TABLE dwd_summary_cars_log, DESCRIPTOR(time_ltz), INTERVAL '3' MINUTES)\n" +
                        ")\n" +
                        "GROUP BY window_start, window_end,cId"
        );


        return table;
    }

    // todo d.创建输出表
    private static void createOutputTable(TableEnvironment tableEnv) {
        // 构建 doris 车辆进出聚合汇总表
        tableEnv.executeSql("CREATE TABLE dws_summary_cars_log (\n" +
                // 窗口开始时间
                "window_start TIMESTAMP(3),\n" +
                // 窗口结束时间
                "window_end TIMESTAMP(3),\n" +
                // 小区ID
                "cId INT,\n" +

                // 驶入次数
                "in_cnt INT,\n" +
                // 驶出次数
                "out_cnt INT\n" +

                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                "'fenodes' = 'node102:8030',\n" +
                "'table.identifier' = 'dws_sca.dws_summary_cars_log',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n" +
                "'sink.label-prefix' = 'doris_cars_log')\n"
        );
    }

    // todo e.保存数据
    private static void saveToSink(TableEnvironment tableEnv, Table table) {
        // todo 创建临时视图 然后写入doris中
        tableEnv.createTemporaryView("dwstable", table);
        //INSERT INTO dws_summary_cars_log
        tableEnv.executeSql(
                "insert into dws_summary_cars_log " +
                        "select " +
                        "window_start,\n" +
                        "window_end,\n" +
                        "cId,\n" +
                        "in_cnt,\n" +
                        "out_cnt \n" +
                        "from dwstable"
        );
    }










}
